package com.mgm.kafka.service.impl;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mgm.advertisement.service.AdvertisementService;
import com.mgm.base.ConverAdRoot;
import com.mgm.constant.GobalConstant;
import com.mgm.creative.service.CreativeService;
import com.mgm.creativeapp.service.CreativeOfAppService;
import com.mgm.custom.applift.AppliftConverToAdvertisement;
import com.mgm.custom.cheetah.CheetahConverToAdvertisement;
import com.mgm.custom.clickdealer.ClickdealerConverToAdvertisement;
import com.mgm.custom.mobpals.MobpalsConverToAdvertisement;
import com.mgm.dto.mobi.MbRoot;
import com.mgm.pojo.Advertisement;
import com.mgm.pojo.Creative;
import com.mgm.pojo.CreativeOfApp;
import com.mgm.utils.FastJsonUtils;
import com.mgm.utils.RandomCodeUtils;

@Service
public class KafkaConsumerSerivceImpl implements MessageListener<String, String> {
	
	private static Logger logger = LoggerFactory.getLogger(KafkaConsumerSerivceImpl.class);
	
	@Autowired
	private AdvertisementService advertisementService;
	@Autowired
	private CreativeService creativeService;
	@Autowired
	private CreativeOfAppService creativeOfAppService;
	
    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
    	logger.debug("【Kafka consume/receive msg】 -- topic:{}   key:{}  val:{}  offset:{}  partition:{}", record.topic(), record.key(), record.value(), record.offset(), record.partition());
    	JSONObject jsonobj = FastJsonUtils.stringToJson(record.value()); //string to json
    	
        //根据不同主题，消费
    	String[] topic = record.topic().split("_"); //前部是广告主ID,后部为表的主键ID
    	String aderCode = topic[0];
    	Integer aderId = Integer.valueOf(topic[1]);
    	
    	if(GobalConstant.CHEETAH_ADV_ID.equals(aderCode)){ //Cheetah猎豹
    		String status = jsonobj.getString("status_code"); //200 for success​/​other code is error 
    		if("200".equals(status)) {
    			JSONObject jobj = jsonobj.getJSONObject("data");
    			JSONArray jarr = jobj.getJSONArray("campaigns");
    			
    			if(jarr.isEmpty()) return;
    			
    			JSONArray ck_jarr = CheetahConverToAdvertisement.checkOffer(jarr); //检查offer
    			ConverAdRoot root = CheetahConverToAdvertisement.toAdvertisement(aderId, ck_jarr);
    			intoDB(root); //入库
    		}
    		
    	} else if (GobalConstant.MOBPALS_ADV_ID.equals(topic[0])){ //Mobpals
    		String status = jsonobj.getString("status"); //Offer status=1 indicate online 
    		if("1".equals(status)) {
    			JSONArray jarr = jsonobj.getJSONArray("data");
    			if(jarr.isEmpty()) return;
    			JSONArray ck_jarr = MobpalsConverToAdvertisement.checkOffer(jarr); //检查offer
    			ConverAdRoot root = MobpalsConverToAdvertisement.toAdvertisement(aderId, ck_jarr);
    			intoDB(root); //入库
    		}
    	} else if (GobalConstant.CLICKDEALER_ADV_ID.equals(topic[0])){ //Clickdealer
    		String status = jsonobj.getString("status");
    		if(status.equals("OK")) { //ok 为该offer active,否则舍弃
    			JSONArray jarr = jsonobj.getJSONArray("ads");
    			if(jarr.isEmpty()) return;
    			JSONArray ck_jarr = ClickdealerConverToAdvertisement.checkOffer(jarr); //检查offer
    			ConverAdRoot root = ClickdealerConverToAdvertisement.toAdvertisement(aderId, ck_jarr);
    			intoDB(root); //入库
    		}
        } else if (GobalConstant.APPLIFT_ADV_ID.equals(topic[0])) { //Applift 
        	String results = jsonobj.getString("results");
        	if(!StringUtils.isEmpty(results)) {
        		JSONArray jarr = jsonobj.getJSONArray("results");
        		if(jarr.isEmpty()) return;
    			JSONArray ck_jarr = AppliftConverToAdvertisement.checkOffer(jarr); //检查offer
    			ConverAdRoot root = AppliftConverToAdvertisement.toAdvertisement(aderId, ck_jarr);
    			intoDB(root); //入库)
        	}
        	
        }
    }
    
    private void intoDB(ConverAdRoot root) {
    	if(root == null || CollectionUtils.isEmpty(root.getAdlist())) return;
    	List<Advertisement> dbList = advertisementService.findByAdvIdAndOldOfferId(root.getAdId(), root.getAdids()); //根据拉取的offer id 去数据库查询一遍
    	advertisementService.updateOfferByAdvIdAndEnabledAndNotInOfferIds(root.getAdId(), root.getAdids()); //更新数据库其他offer状态和计数器
    	Date curTime = new Date();
    	List<Advertisement> adlist = root.getAdlist();
    	Map<String, Advertisement> admap = root.getAdmap();
    	Map<String, Creative> ctmap = root.getCtmap();
    	Map<String, CreativeOfApp> appmap = root.getAppmap();
    	
    	List<Creative> insertctList = new ArrayList<>(); //素材数组
    	List<CreativeOfApp> insertappList = new ArrayList<>(); //商店素材数组
    	if(CollectionUtils.isEmpty(dbList)) { //如果数据库没有，全部入库
    		Creative ct = null;
    		CreativeOfApp app = null;
    		for (Advertisement advertisement : adlist) {
    			String offerid = advertisement.getP1() + RandomCodeUtils.randomNumeric(8); //系统生成offerid
    			advertisement.setOfferId(offerid);
    			if(ctmap != null) {
    				ct = ctmap.get(advertisement.getP1());
    				ct.setOfferid(offerid);
    				insertctList.add(ct);
    			}
    			if(appmap != null) {
    				app = appmap.get(advertisement.getP1());
    				app.setOfferid(offerid);
    				insertappList.add(app);
    			}
			}
    		int i = advertisementService.insertBatch(adlist);
    		
        	logger.info("================[insert]===1.成功插入offer["+i+"]条==================");
    	} else { //部分入库，部分更新
    		List<Advertisement> updateadmentList = new ArrayList<>(); //需更新的offer
    		Advertisement ad = null;
    		Creative ct = null;
    		CreativeOfApp app = null;
    		for (Advertisement db_ad : dbList) { //遍历数据库中offer
    			ad = admap.get(db_ad.getP1());
    			ad.setOfferId(db_ad.getOfferId());
    			ad.setStart_at(db_ad.getStart_at());
    			ad.setUpdate_time(curTime);
    			if(db_ad.getEnabled()==false) { //上游offer开启，本地offer关闭，更新状态和更新计数
    				ad.setEnd_at(null);
    				ad.setCounter(db_ad.getCounter()+1);
	    		}
    			updateadmentList.add(ad); //更新為上游數據
    			
    			if(ctmap != null) {
    				ct = ctmap.get(ad.getP1());
    				ct.setOfferid(ad.getOfferId());
    				insertctList.add(ct);
    			}
    			if(appmap != null) {
    				app = appmap.get(ad.getP1());
    				app.setOfferid(ad.getOfferId());
    				insertappList.add(app);
    			}
    			
    			admap.remove(db_ad.getP1());
			}
    		if(!CollectionUtils.isEmpty(updateadmentList)) {
    			int i = advertisementService.updateBatch(updateadmentList);
    			logger.info("================[update]===成功更新offer["+i+"]条==================");
    		}
    		
    		//map中不为空，为其做插入
    		Iterator<Map.Entry<String, Advertisement>> it = admap.entrySet().iterator();  
    		List<Advertisement> insertadmentList = new ArrayList<>();
    		Advertisement adment = null;
    		Creative creat = null;
    		CreativeOfApp appcreat = null;
            while(it.hasNext()){
				Map.Entry<String, Advertisement> entry = it.next();
				adment = entry.getValue();
        		String offerid = adment.getP1() + RandomCodeUtils.randomNumeric(8); //系统生成offerid
        		adment.setOfferId(offerid);
        		
    			insertadmentList.add(adment); //加入list
    			
    			if(ctmap != null) {
    				creat = ctmap.get(adment.getP1());
    				creat.setOfferid(adment.getOfferId());
    				insertctList.add(creat);
    			}
    			if(appmap != null) {
    				appcreat = appmap.get(adment.getP1());
    				appcreat.setOfferid(adment.getOfferId());
    				insertappList.add(appcreat);
    			}
    			
            }
            if(!CollectionUtils.isEmpty(insertadmentList)) {
            	int i = advertisementService.insertBatch(insertadmentList);
            	logger.info("================[insert]===2.成功插入offer["+i+"]条==================");
            }
    	}
    	
    	//先删除当前拉取offer相关的素材
    	creativeService.deleteByCreative(insertctList); //删除app素材，重新入库
    	if(!CollectionUtils.isEmpty(insertctList)) { //素材入库
    		int i = creativeService.insertBatch(insertctList);
    		logger.info("================[insert]===3.成功插入素材["+i+"]条==================");
    	}
    	
    	//先删除当前拉取offer相关的app素材
    	creativeOfAppService.deleteByCreativeOfApp(insertappList); //删除app素材，重新入库
    	if(!CollectionUtils.isEmpty(insertappList)) { //app素材入库
    		int i = creativeOfAppService.insertBatch(insertappList);
    		logger.info("================[insert]===4.成功插入app素材["+i+"]条==================");
    	}
    	 
    }
    
    /**
     * 
     * @Title: verifyMbOffer   
     * @Description: https://appwalls.mobi/api 接口参数规则过滤   
     * @param: @param mbRoot
     * @param: @return      
     * @return: boolean      
     * @throws
     */
    private boolean verifyMbOffer(MbRoot mbRoot) {
    	if(mbRoot.getOfferid() == null || "".equals(mbRoot.getOfferid())) return false;
    	if(mbRoot.getTitle() == null || "".equals(mbRoot.getTitle())) return false;
    	if(mbRoot.getClickurl() == null || "".equals(mbRoot.getClickurl())) return false;
    	if(mbRoot.getPayout_type() == null || "".equals(mbRoot.getPayout_type())) return false;
    	if(mbRoot.getPayout()<=0) return false;
    	if(mbRoot.getCountries() == null || "".equals(mbRoot.getCountries())) return false;
    	if(mbRoot.getAppid() == null || "".equals(mbRoot.getAppid())) return false;
//    	if(mbRoot.getDevice_types() == null || mbRoot.getDevice_types().isEmpty()) return false; //连接上有os，和devicetype（三星，华为之类的）不一样
    	
    	return true;
    }
    
    private void mbRootIntoDB(Integer advId, List<String> offerIds, Map<String, MbRoot> map) {
    	Date curTime = new Date();
    	DateTime jodaTime = new DateTime(curTime);
    	DateTime n = jodaTime.plusHours(2);
    	
    	List<Advertisement> dbList = advertisementService.findByAdvIdAndOldOfferId(advId, offerIds); //根据拉取的offer id 去数据库查询一遍
    	int c = advertisementService.updateOfferByAdvIdAndEnabledAndNotInOfferIds(advId, offerIds); //更新数据库其他offer状态和计数器
    	
    	List<CreativeOfApp> cofAppList = new ArrayList<>(); //商店素材数组
    	CreativeOfApp cofApp = null;
    	if(CollectionUtils.isEmpty(dbList)) { //如果数据库没有，全部入库
    		List<Advertisement> admentList = new ArrayList<>(); //广告数组
        	Advertisement adment = null; //广告
        	for (MbRoot mbRoot : map.values()) {
        		adment = new Advertisement();
        		cofApp = new CreativeOfApp();
        		adment.setAdv_id(advId); 
        		adment.setExpire_date(n.toDate());
        		adment = mbRootToAdvertisement(mbRoot, adment, curTime, true);
        		admentList.add(adment); //加入offer list
        		cofApp = mbRootToCreativeOfApp(mbRoot, cofApp, curTime);
        		cofApp.setOfferid(adment.getOfferId());
        		cofAppList.add(cofApp); //加入app素材 list
    		}
        	
        	int i = advertisementService.insertBatch(admentList);
        	logger.info("================[insert]===1.成功插入offer["+i+"]条==================");
    	} else { //部分入库，部分更新
    		List<Advertisement> updateadmentList = new ArrayList<>(); //需更新的offer
    		for (Advertisement advertisement : dbList) { //遍历数据库中offer
				cofApp = new CreativeOfApp(); //todo...
				MbRoot mb = map.get(advertisement.getP1());
				advertisement = mbRootToAdvertisement(mb, advertisement, curTime, false);
				if(advertisement.getEnabled()==false) { //上游offer开启，本地offer关闭，更新状态和更新计数
	    			advertisement.setEnabled(true);
	    			advertisement.setCounter(advertisement.getCounter()+1);
	    		}
    			updateadmentList.add(advertisement); //Advertisement对象加入批量更新list
    			
    			cofApp = mbRootToCreativeOfApp(mb, cofApp, curTime);
        		cofApp.setOfferid(advertisement.getOfferId());
        		cofAppList.add(cofApp); //加入app素材 list
    			
    			map.remove(advertisement.getP1()); //上游存在的offer更新完数据库，删除map，下面为没删除的做插入
    			
			}
    		if(!CollectionUtils.isEmpty(updateadmentList)) {
    			int i = advertisementService.updateBatch(updateadmentList);
    			logger.info("================[update]===成功更新offer["+i+"]条==================");
    		}
    		
    		//map中不为空，为其做插入
    		Iterator<Map.Entry<String, MbRoot>> it = map.entrySet().iterator();  
    		List<Advertisement> insertadmentList = new ArrayList<>();
    		Advertisement adment = null;
            while(it.hasNext()){
				Map.Entry<String, MbRoot> entry = it.next();
				adment = new Advertisement();
        		adment.setAdv_id(advId); 
        		adment.setExpire_date(n.toDate());
        		adment = mbRootToAdvertisement(entry.getValue(), adment, curTime, true);
        		insertadmentList.add(adment); //加入list
        		
        		cofApp = mbRootToCreativeOfApp(entry.getValue(), cofApp, curTime);
        		cofApp.setOfferid(adment.getOfferId());
        		cofAppList.add(cofApp); //加入app素材 list
            }
            if(!CollectionUtils.isEmpty(insertadmentList)) {
            	int i = advertisementService.insertBatch(insertadmentList);
            	logger.info("================[insert]===2.成功插入offer["+i+"]条==================");
            }
    	}
    	
    	//先删除当前拉取offer相关的app素材
    	creativeOfAppService.deleteByCreativeOfApp(cofAppList); //删除app素材，重新入库
    	if(!CollectionUtils.isEmpty(cofAppList)) { //app素材入库
    		int i = creativeOfAppService.insertBatch(cofAppList);
    		logger.info("================[insert]===3.成功插入app素材["+i+"]条==================");
    	}
    	
    }
    
    /**
     * 将json模型MbRoot转为Advertisement对象
     * @Title: mbRootToAdvertisement   
     * @Description: TODO(...)   
     * @param: @param mbRoot
     * @param: @param advertisement
     * @param: @param curTime 当前时间
     * @param: @param isInsert true：新增；false：更新
     * @param: @return      
     * @return: Advertisement      
     * @throws
     */
    private Advertisement mbRootToAdvertisement(MbRoot mbRoot, Advertisement advertisement, Date curTime, boolean isInsert) {
    	if(isInsert) { //新增
    		advertisement.setP1(mbRoot.getOfferid()+""); //备份上游offerid
    		advertisement.setOfferId(mbRoot.getOfferid()+RandomCodeUtils.randomNumeric(8));//生成offerid
    		advertisement.setOfferName(mbRoot.getTitle());
    		advertisement.setClick_url(mbRoot.getClickurl());
    		advertisement.setPay_price(BigDecimal.valueOf(mbRoot.getPayout()));
    		advertisement.setPay_type(mbRoot.getPayout_type());
    		advertisement.setCountries(mbRoot.getCountries());
    		advertisement.setExcitation("no".equals(mbRoot.getIncent())?false:true); //是否为激励广告
    		advertisement.setRequires("no".equals(mbRoot.getGaid_idfa_require()) ? null : mbRoot.getGaid_idfa_require()); //设备id
    		
    		advertisement.setAppname(mbRoot.getTitle());
    		advertisement.setOs("Android");
    		advertisement.setPkgname(mbRoot.getAppid());
    		advertisement.setIcon(mbRoot.getIcon());
    		advertisement.setStart_at(curTime);
    		advertisement.setEnabled(true);
    		advertisement.setCurrency("USD"); //默认USD
    		advertisement.setCounter(0); //插入默认0
    	} else { //更新
    		advertisement.setOfferName(mbRoot.getTitle());
    		advertisement.setClick_url(mbRoot.getClickurl());
    		advertisement.setPay_price(BigDecimal.valueOf(mbRoot.getPayout()));
    		advertisement.setPay_type(mbRoot.getPayout_type());
    		advertisement.setCountries(mbRoot.getCountries());
    		advertisement.setExcitation("no".equals(mbRoot.getIncent())?false:true); //是否为激励广告
    		advertisement.setRequires("no".equals(mbRoot.getGaid_idfa_require()) ? null : mbRoot.getGaid_idfa_require()); //设备id
    		
    		advertisement.setAppname(mbRoot.getTitle());
    		advertisement.setOs("Android");
    		advertisement.setPkgname(mbRoot.getAppid());
    		advertisement.setIcon(mbRoot.getIcon());
    		//状态和计数器在上级判断和处理
    		advertisement.setUpdate_time(curTime);
    	}
    	return advertisement;
    }
    
    /**
     * 
     * @Title: mbRootToCreativeOfApp   
     * @Description: TODO(...)   
     * @param: @param mbRoot
     * @param: @param creativeOfApp
     * @param: @param curTime
     * @param: @return      
     * @return: CreativeOfApp      
     * @throws
     */
    private CreativeOfApp mbRootToCreativeOfApp(MbRoot mbRoot, CreativeOfApp creativeOfApp, Date curTime) {
    	creativeOfApp.setAppname(mbRoot.getTitle());
    	creativeOfApp.setCountry(mbRoot.getCountries());
//    	creativeOfApp.setLanguage("");
//    	creativeOfApp.setCurVersion(curVersion);
    	creativeOfApp.setOsVersion(mbRoot.getInclude_min_os_version().get(0));
    	creativeOfApp.setDescription(mbRoot.getDescription());
    	creativeOfApp.setIconurl(mbRoot.getIcon());
    	creativeOfApp.setStoreRating(mbRoot.getApprating()+"");
    	creativeOfApp.setUpdateTime(curTime);
    	creativeOfApp.setEnabled(true);
    	return creativeOfApp;
    }
    
    
}